依赖注入

本质就是把一段需要复用的代码作为依赖项,注入到需要操作的函数中

看起来和装饰器没什么区别,在python中依赖项只要是能调用的即可,那么及时是类也能注入了

“依赖注入”非常适用于以下使用场景:

1、业务逻辑复用

2、共享数据库连接

3、安全机制、权限校验、角色管理等等

4、后台任务注入

所有上述使用场景,借助于”依赖注入”可以明确的提高代码复用,减少代码重复

fastapi简单示例

def common():
    return {"a":1,"b":2}

@app.get("/item")
def read_common(commons:dict = Depends(common)):
    # 依赖项common注入到read_common中,执行结果作为参数返回给commons
    return commons

tips:Depends和body,path属于同一类,具有相似的属性操作

执行流程

  • 调用依赖函数
  • 依赖函数的结果作为参数传入到被依赖函数中

利用依赖注入实现简易分页器

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

依赖子项(依赖套娃)

依赖项也可以被注入,具体层级不做限制,若一个依赖项被多次注入到同一个请求中,fastapi默认只会执行一次来缓存这个函数的结果,再次调用从缓存中拿,若我们还是需要调用,可以用Depends的参数*use_cache=False*来禁止依赖项的缓存

示例:

def query_extractor(q11: str, q12: str):
    return q11 + q12


def query_or_cookie_extractor(q2: str = Depends(query_extractor), last_query: str = Cookie(None)):
    if not q2:
        return last_query
    return q2


@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
return {"q_or_cookie": query_or_default}

基于路径装饰器的依赖项

路径装饰器有一个dependencies参数,当我们需要依赖注入但是不需要返回结果的时候即可在这里使用

async def verify_token(x_token: str = Header(...)):
    if x_token != "token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

依赖项的参数的顺序是固定的

基于上下文管理器和依赖注入实现数据库资源管理

示例

class DBSession():
    pass


class MySuperContextManager:
    def __init__(self):
        print("创建数据库对象")
        self.db = DBSession()

    def __enter__(self):
        print("使用数据库对象")
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        print("关闭数据库对象")

async def get_db():
    with MySuperContextManager() as db:
        yield db

@app.get("/items/")
async def read_items(db:MySuperContextManager = Depends(get_db)):
    print(db)
    return [{"item": "Foo"}, {"item": "Bar"}]

可参数化的依赖项

示例


class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
      # 这里的q是通过查询参数调用进来的
        if q:
            return self.fixed_content in q
        return False


checker = FixedContentQueryChecker("bar")

@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}